package com.ikas.ai.server.kafka;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONValidator;
import com.ikas.ai.server.module.alarm.service.UnitDeteriorationAlarmService;
import com.ikas.ai.server.module.assessment.service.AssessModelService;
import com.ikas.ai.server.module.condition.dao.WorkConditionFlowDao;
import com.ikas.ai.server.module.condition.dao.WorkConditionStepDetailDao;
import com.ikas.ai.server.module.condition.model.WorkConditionFlow;
import com.ikas.ai.server.module.condition.model.WorkConditionStepDetail;
import com.ikas.ai.server.module.instance.dorisDao.DorisWorkInstanceStepDetailDao;
import com.ikas.ai.server.module.instance.model.DorisWorkFlowInstance;
import com.ikas.ai.server.module.instance.model.DorisWorkInstanceStepDetail;
import com.ikas.ai.server.module.instance.model.dto.DorisWorkInstanceDetailHisDTO;
import com.ikas.ai.server.module.instance.service.DorisWorkFlowInstanceService;
import com.ikas.ai.server.module.instance.service.DorisWorkInstanceStepDetailCostService;
import com.ikas.ai.server.module.instance.service.DorisWorkInstanceStepDetailService;
import com.ikas.ai.utils.ObjectUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;

/**
 * @desccription 计算完工况测点响应时间长后
 * 1、计算工况平均时长历史（100条可修改），修改t_work_instance_step_detail平均时长字段
 * 2、计算IQR值，如满足条件
 * 2.1 生成劣化预警
 * 2.2 触发评价模型
 * @date 2024-4-12 10:06:37
 */
@Slf4j
@Component
public class WorkFlowInstanceStepDetailAvgCostTimeConsumer {

    @Autowired
    DorisWorkInstanceStepDetailService dorisWorkInstanceStepDetailService;
    @Autowired
    UnitDeteriorationAlarmService unitDeteriorationAlarmService;
    @Autowired
    AssessModelService assessModelService;
    @Autowired
    DorisWorkInstanceStepDetailDao dorisWorkInstanceStepDetailDao;
    @Autowired
    WorkConditionStepDetailDao workConditionStepDetailDao;
    @Autowired
    WorkConditionFlowDao workConditionFlowDao;

    @Autowired
    DorisWorkFlowInstanceService workFlowInstanceService;
    @Autowired
    DorisWorkInstanceStepDetailCostService workInstanceStepDetailCostService;

    @Value("${access.alarm.state}")
    private boolean state;

    //IQR数据查询
    @Value("${work-instance.step-detail.cost-time.max-length:100}")
    private Integer count;


    @KafkaListener(topics = KafkaInitialConfiguration.WORK_INSTANCE_DETAIL_DELAY, groupId = KafkaInitialConfiguration.WORK_INSTANCE_DETAIL_DELAY_GROUP, topicPattern = "0")
    private void processKafkaMessage(List<ConsumerRecord<String, String>> list, Acknowledgment ack) {
        Instant start = Instant.now();
        list.forEach(record -> {
            try {
                List<DorisWorkInstanceStepDetail> detailList = new ArrayList<>();
                final JSONValidator.Type type = JSONValidator.from(record.value()).getType();
                if (type == JSONValidator.Type.Array) {
                    detailList = JSONArray.parseArray(record.value(), DorisWorkInstanceStepDetail.class);
                } else if (type == JSONValidator.Type.Object) {
                    detailList = Collections.singletonList(JSONObject.parseObject(record.value(), DorisWorkInstanceStepDetail.class));
                }
                if (CollectionUtil.isEmpty(detailList)) {
                    return;
                }
                log.info("====STEP_DETAIL-KAFKA-====>offset>[{}]--timestamp>[{}]------>meteDataList:{}", record.offset()
                        , record.timestamp(), JSONUtil.toJsonStr(detailList));
                detailList.forEach(this::processInstanceDetailDelay);
            } catch (Exception e) {
                log.error("=====劣化预警和状态评价检查器：消费错误======", e);
            }
        });
        //手动提交
        ack.acknowledge();
        Instant end = Instant.now();
        log.info("STEP_DETAIL-KAFKA消费执行时长毫秒：{}", Duration.between(start, end).toMillis());
    }

    /**
     * 刷新StepDetail平均耗时表
     *
     * @param detail
     */
    public void generateStepDetailAvgCostTime(DorisWorkInstanceStepDetail detail) {
        DorisWorkFlowInstance flowInstance = workFlowInstanceService.getById(detail.getWfInstanceId());
        if (!ObjectUtil.isNull(flowInstance)) {
            workInstanceStepDetailCostService.computeAvgCostTime(flowInstance.getMachineNo(), detail);
        }
    }

    /**
     * 检查是否符合劣化预警以及状态评价的触发条件
     * @param detail
     */
    public void processInstanceDetailDelay(DorisWorkInstanceStepDetail detail) {

        try {
            //工况详情平均时长
            generateStepDetailAvgCostTime(detail);
        } catch (Exception e) {
            log.error("修改工况详情平均时长错误",e);
        }
        try {
            //劣化预警和触发评价模型
            calculateIQR(detail);
        } catch (Exception e) {
            log.error("劣化预警和触发评价模型错误",e);
        }


    }

    private void calculateIQR(DorisWorkInstanceStepDetail detail) {
        log.info("access.alarm.state评价模型state:{}", state);
        if (state) {
            wakeUpAssessAndUnitDeterioration(detail);
        } else {
            WorkConditionStepDetail workConditionStepDetail = workConditionStepDetailDao.selectById(detail.getDetailId());
            if (workConditionStepDetail == null) {
                log.info("不存在该工况流程测点详情:{}", detail.getDetailId());
                return;
            }
            WorkConditionFlow workConditionFlow = workConditionFlowDao.selectById(workConditionStepDetail.getWfId());
            DorisWorkInstanceDetailHisDTO dorisWorkInstanceDetailHisDTO = new DorisWorkInstanceDetailHisDTO();
            dorisWorkInstanceDetailHisDTO.setMachineNo(workConditionFlow.getMachineNo());
            dorisWorkInstanceDetailHisDTO.setWfName(workConditionFlow.getWfName());
            dorisWorkInstanceDetailHisDTO.setMeteCode(workConditionStepDetail.getMeteCode());
            dorisWorkInstanceDetailHisDTO.setCount(count);
            List<Double> list = dorisWorkInstanceStepDetailDao.getHisData(dorisWorkInstanceDetailHisDTO);
            if (CollectionUtil.isEmpty(list)) {
                log.info("detailList is null");
                return;
            }
            double[] hisData = list.stream().filter(Objects::nonNull).mapToDouble(i -> i).toArray();
            if (hisData == null || detail.getCostTime() == null || hisData.length == 0) {
                if (hisData == null) {
                    log.info("hisData == null");
                } else if (detail.getCostTime() == null) {
                    log.info("detail.getCostTime()==null");
                } else if (hisData.length == 0) {
                    log.info("hisData.length==0");
                }
                return;
            }
            if (calculateIQRByCommonMath3(hisData, detail.getCostTime().doubleValue())) {
                wakeUpAssessAndUnitDeterioration(detail);
            }
        }
    }

    public void wakeUpAssessAndUnitDeterioration(DorisWorkInstanceStepDetail detail) {
        try {
            //触发工况劣化预警
            unitDeteriorationAlarmService.add(detail);
        } catch (Exception e) {
            log.error(e.getMessage());
        }
        try {
            //触发评价模型
            assessModelService.wakeAssessByInstanceDetail(detail);
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    private boolean calculateIQRByCommonMath3(double[] hisData, double data) {
        DescriptiveStatistics stats = new DescriptiveStatistics(hisData);
        double q1 = stats.getPercentile(25); // 第一四分位数（Q1）
        double q3 = stats.getPercentile(75); // 第三四分位数（Q3）
        double iqr = q3 - q1; // 四分位数范围（IQR）
        // 计算上须和下须
//        double lowerWhisker = q1 - 1.5 * iqr;
        double upperWhisker = q3 + 1.5 * iqr;
        log.info("=====IRQ======上须值计算：hisData:{},data:{},upperWhisker:{},{}", hisData, data, upperWhisker, data > upperWhisker);
        return data > upperWhisker;
    }

    /**
     * 获取0点的时间戳
     *
     * @param daysAgo
     * @return
     */
    public static Date getStartOfDayTimestampAgo(int daysAgo) {
        LocalDateTime dateTime = LocalDateTime.now().minusDays(daysAgo);
        LocalDateTime startOfDay = dateTime.withHour(0).withMinute(0).withSecond(0).withNano(0);
        return Date.from(startOfDay.atZone(ZoneId.systemDefault()).toInstant());
    }

    public static Date getStartOfDayTimestampAfter(int dayAfter) {
        LocalDateTime dateTime = LocalDateTime.now().plusDays(dayAfter);
        LocalDateTime startOfDay = dateTime.withHour(0).withMinute(0).withSecond(0).withNano(0);
        return Date.from(startOfDay.atZone(ZoneId.systemDefault()).toInstant());
    }
}
